package relop;

/**
 * The simplest of all join algorithms: nested loops (see textbook, 3rd edition,
 * section 14.4.1, page 454).
 */
public class SimpleJoin extends Iterator {

	private Iterator left;
	private Iterator right;
	private Predicate[] preds;
	private Tuple leftTuple;
	private Tuple nextTuple;
	
	/**
	 * Constructs a join, given the left and right iterators and join predicates
	 * (relative to the combined schema).
	 */
	public SimpleJoin(Iterator left, Iterator right, Predicate... preds) {
		this.left = left;
		this.right = right;
		this.schema = Schema.join(left.schema, right.schema);
		this.preds = preds;
		leftTuple = null;
		nextTuple = null;
	}

	/**
	 * Gives a one-line explaination of the iterator, repeats the call on any
	 * child iterators, and increases the indent depth along the way.
	 */
	public void explain(int depth) {
		indent(depth);
		String explaination = "SimpleJoin with predicates: ";
		for(int i = 0; i < preds.length; i++) {
			if(i == 0)
				explaination += preds[i].toString();
			else
				explaination += " OR " + preds[i].toString();
		}
		System.out.println(explaination);
		left.explain(depth + 1);
		right.explain(depth + 1);
	}

	/**
	 * Restarts the iterator, i.e. as if it were just constructed.
	 */
	public void restart() {
		left.restart();
		right.restart();
		leftTuple = null;
		nextTuple = null;
	}

	/**
	 * Returns true if the iterator is open; false otherwise.
	 */
	public boolean isOpen() {
		return (left.isOpen() && right.isOpen());
	}

	/**
	 * Closes the iterator, releasing any resources (i.e. pinned pages).
	 */
	public void close() {
		left.close();
		right.close();
		leftTuple = null;
		nextTuple = null;
	}

	/**
	 * Returns true if there are more tuples, false otherwise.
	 */
	public boolean hasNext() {
		if(!isOpen())
			return false;
		
		while(true) {
			//This loop only exit if found a good tuple or the left depleted.
			
			//Fetch a tuple in the left scan if we do not already have one
			if(leftTuple == null) {
				if(!left.hasNext()) {
					nextTuple = null;
					return false;	//Left depleted, nothing to join
				}
				
				leftTuple = left.getNext();
			}
		
			//Loop on tuples in the right scan
			while(right.hasNext()) {
				Tuple rightTuple = right.getNext();
				Tuple tmpTuple = Tuple.join(leftTuple, rightTuple, schema);
				
				boolean pass = false;
				
				for(Predicate pred : preds) {
					if(pred.evaluate(tmpTuple)) {
						pass = true;
						break;
					}
				}
				
				if(pass) {
					nextTuple = tmpTuple;
					return true;
				}
			}
			
			//Right depleted, restart
			leftTuple = null;	//Signal the left to fetch new tuple
			right.restart();
		}
	}

	/**
	 * Gets the next tuple in the iteration.
	 * 
	 * @throws IllegalStateException
	 *             if no more tuples
	 */
	public Tuple getNext() {
		if(nextTuple == null)
			throw new IllegalStateException("getNext(): no more tuples");
		
		return nextTuple;
	}

} // public class SimpleJoin extends Iterator
